package threadpool;

import com.sun.org.apache.bcel.internal.generic.NEW;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ClassName: PauseableThreadPool
 * Package: threadpool
 *
 * @Author 马学兴
 * @Create 2024/3/14 20:35
 * @Version 1.0
 * Description:演示每个任务执行的前后，放钩子函数
 */
public class PauseableThreadPool extends ThreadPoolExecutor {
    private final ReentrantLock lock = new ReentrantLock(); // 用于同步的锁对象
    private Condition unpaused = lock.newCondition(); // 用于线程暂停和恢复的条件变量
    private boolean isPaused; // 标志线程池是否被暂停

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    // 在执行任务前进行钩子方法的拦截
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        lock.lock(); // 获取锁
        try {
            while (isPaused) { // 如果线程池被暂停，则等待
                unpaused.await();
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock(); // 释放锁
        }
    }

    // 暂停线程池
    private void pause() {
        lock.lock(); // 获取锁
        try {
            isPaused = true; // 设置暂停标志为 true
        } finally {
            lock.unlock(); // 释放锁
        }
    }

    // 恢复线程池
    public void resume(){
        lock.lock(); // 获取锁
        try {
            isPaused=false; // 设置暂停标志为 false
            unpaused.signalAll(); // 唤醒所有等待线程
        }finally {
            lock.unlock(); // 释放锁
        }
    }

    // 测试方法
    public static void main(String[] args) throws InterruptedException {
        PauseableThreadPool pauseableThreadPool = new PauseableThreadPool(10, 20, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                System.out.println("我被执行");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        };
        for (int i = 0; i < 1000; i++) {
            pauseableThreadPool.execute(runnable); // 提交任务给线程池执行
        }

        Thread.sleep(1500); // 等待一段时间
        pauseableThreadPool.pause(); // 暂停线程池
        System.out.println("线程池被暂停了");
        Thread.sleep(1500); // 继续等待一段时间
        pauseableThreadPool.resume(); // 恢复线程池
        System.out.println("线程池被恢复了");
    }
}
